package study.mq.kafka.consumer.listener._2_offset;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.springframework.stereotype.Component;

import java.util.Map;

/*（★）异步提交
 * （1）一种非阻塞式方法，即无需等待当前消息是否提交成功或者失败，就会继续下一步操作
 * （2）对于失败或者成功后续的处理，可以在定义的回调函数中处理commitAsync(callback)
 */
@Slf4j
@Component
public class _3_CommitListener {

//    @KafkaListener(id = "OFFSET", topics = "_2_offset-topic")
    public void onMessage(ConsumerRecord<String, String> record, Consumer consumer) {
        String topic = record.topic();
        int partition = record.partition();
        long offset = record.offset();
        log.info("record _2_offset: {}", offset);



        consumer.commitAsync();
        consumer.commitAsync(new OffsetCommitCallback() {
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                log.info("");
            }
        });
//        consumer.commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);
    }
}
